package org.truenewx.tnxjee.core.util.concurrent;

import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.truenewx.tnxjee.core.util.LogUtil;
import org.truenewx.tnxjee.core.util.ThreadUtil;

import java.util.Collection;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * 线程池工具类
 *
 * @author jianglei
 */
public class ExecutorUtil {

    private ExecutorUtil() {
    }

    public static final String DEFAULT_EXECUTOR_BEAN_NAME = "defaultExecutor";
    public static final String SCHEDULED_EXECUTOR_BEAN_NAME = "scheduledExecutor";

    public static ExecutorService buildDefaultExecutor(int corePoolSize) {
        return new DefaultThreadPoolExecutor(corePoolSize);
    }

    public static ScheduledExecutorService buildScheduledExecutor(int corePoolSize) {
        BasicThreadFactory factory = new BasicThreadFactory.Builder()
                .namingPattern(ScheduledExecutorService.class.getSimpleName() + "-%d")
                .daemon(true).build();
        return new ScheduledThreadPoolExecutor(corePoolSize, factory);
    }

    @SafeVarargs
    public static <R> R retryUtilTimeout(long interval, long timeout, Supplier<R> supplier,
            Class<? extends Throwable>... causes) {
        if (interval < 0) {
            interval = 0;
        }
        R result = null;
        long time0 = System.currentTimeMillis();
        while (result == null) {
            try {
                result = supplier.get();
            } catch (Exception e) {
                if (isNotCause(e, causes)) {
                    throw e;
                }
                // 如果超时则抛出异常
                if (timeout > interval && (System.currentTimeMillis() - time0) >= timeout) {
                    throw e;
                }
                LogUtil.warn(supplier.getClass(), e.getMessage());
                LogUtil.warn(supplier.getClass(), "Retry in " + interval + "ms.");
                ThreadUtil.sleep(interval);
            }
        }
        return result;
    }

    private static boolean isNotCause(Exception e, Class<? extends Throwable>[] causes) {
        if (causes.length == 0) {
            return false;
        }
        for (Class<? extends Throwable> cause : causes) {
            if (cause.isAssignableFrom(e.getClass())) {
                return false;
            }
        }
        return true;
    }

    @SafeVarargs
    public static <R> R retryUtilSuccessful(Supplier<R> supplier, Class<? extends Throwable>... causes) {
        return retryUtilSuccessful(supplier, e -> {
            LogUtil.warn(supplier.getClass(), e.getMessage());
        }, causes);
    }

    @SafeVarargs
    public static <R> R retryUtilSuccessful(Supplier<R> supplier, Consumer<Exception> exceptionHandler,
            Class<? extends Throwable>... causes) {
        while (true) {
            try {
                return supplier.get();
            } catch (Exception e) {
                if (isNotCause(e, causes)) {
                    throw e;
                }
                exceptionHandler.accept(e);
                long interval = 100;
                LogUtil.warn(supplier.getClass(), "Retry in " + interval + "ms.");
                ThreadUtil.sleep(interval);
            }
        }
    }

    /**
     * 异步映射
     *
     * @param sources  源集合
     * @param mapper   映射函数，如果返回结果为null，则不加入结果集合
     * @param executor 线程池
     * @param <T>      源类型
     * @param <R>      映射类型
     * @return 映射后的集合
     */
    public static <T, R> List<R> asyncMap(Collection<T> sources, Function<T, R> mapper, Executor executor) {
        return sources.stream().map(source -> CompletableFuture.supplyAsync(() -> mapper.apply(source), executor))
                .map(CompletableFuture::join).filter(Objects::nonNull).toList();
    }

}
